package pbservice

import "net"
import "fmt"
import "net/rpc"
import "log"
import "time"
import "viewservice"
import "sync"
import "sync/atomic"
import "os"
import "syscall"
import (
	"math/rand"
)

type PBServer struct {
	mu         sync.Mutex
	l          net.Listener
	dead       int32 // for testing
	unreliable int32 // for testing
	me         string
	vs         *viewservice.Clerk
	// Your declarations here.

	Viewnum uint
	Primary string
	Backup  string

	data       map[string]string
	requestMap map[int64]bool
	dataLocker *sync.RWMutex
}

func (pb *PBServer) Get(args *GetArgs, reply *GetReply) error {

	// Your code here.
	if pb.me == pb.Primary {
		pb.dataLocker.RLock()
		if value, ok := pb.data[args.Key]; ok {
			reply.Value = value
		} else {
			reply.Value = ""
		}
		pb.dataLocker.RUnlock()
		reply.Viewnum = pb.Viewnum
	} else {
		var primaryReply GetReply
		call(pb.Backup, "PBServer.Get", args, &primaryReply)
		reply.Value = primaryReply.Value
		reply.Viewnum = primaryReply.Viewnum
	}
	return nil
}

func (pb *PBServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) error {

	// Your code here.

	if pb.isdead() {
		for {
			time.Sleep(viewservice.PingInterval)
			view, err := pb.vs.Get()

			if !err {
				continue
			}
			if view.Viewnum != pb.Viewnum {
				pb.Viewnum = view.Viewnum
				break
			}
		}
	}

	if pb.me != pb.Primary && !args.FromPrimary {
		call(pb.Primary, "PBServer.PutAppend", args, reply)
		return nil
	}
	pb.dataLocker.Lock()
	defer pb.dataLocker.Unlock()
	if _, ok := pb.requestMap[args.Number]; ok {
		return nil
	}

	if args.IsPut {
		pb.data[args.Key] = args.Value
	} else {
		pb.data[args.Key] += args.Value
	}
	pb.requestMap[args.Number] = true

	if pb.me == pb.Primary && pb.Backup != "" {
		var backupReply PutAppendReply
		args.FromPrimary = true
		args.IsPut = true
		args.Value = pb.data[args.Key]
		call(pb.Backup, "PBServer.PutAppend", args, &backupReply)
	}
	reply.Viewnum = pb.Viewnum
	return nil
}

func (pb *PBServer) UpdateData(args *UpdateDataArgs, reply *UpdateDataArgs) error {
	pb.dataLocker.Lock()
	defer pb.dataLocker.Unlock()
	for k, _ := range pb.data {
		delete(pb.data, k)
	}

	for k, v := range args.Data {
		pb.data[k] = v
	}

	for k, _ := range pb.requestMap {
		delete(pb.requestMap, k)
	}

	for k, v := range args.RequestMap {
		pb.requestMap[k] = v
	}

	return nil
}

//
// ping the viewserver periodically.
// if view changed:
//   transition to new view.
//   manage transfer of state from primary to new backup.
//
func (pb *PBServer) tick() {

	// Your code here.
	view, err := pb.vs.Ping(pb.Viewnum)
	if err != nil {
		pb.Viewnum = 0
		return
	}

	pb.Viewnum = view.Viewnum
	pb.Primary = view.Primary

	// when found a backup send all data to backup
	if pb.me == view.Primary && pb.Backup != view.Backup && view.Backup != "" {
		pb.dataLocker.RLock()
		args := &UpdateDataArgs{Data: pb.data, RequestMap: pb.requestMap}
		reply := UpdateDataReply{}
		if call(view.Backup, "PBServer.UpdateData", args, &reply) {
			pb.Backup = view.Backup
		} else {
			pb.Backup = ""
		}
		pb.dataLocker.RUnlock()
	} else {
		pb.Backup = view.Backup
	}

}

func (pb *PBServer) call(srv string, rpcname string, args interface{}, reply interface{}) bool {
	c, errx := rpc.Dial("unix", srv)
	if errx != nil {
		return false
	}
	defer c.Close()

	err := c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	return false
}

// tell the server to shut itself down.
// please do not change these two functions.
func (pb *PBServer) kill() {
	atomic.StoreInt32(&pb.dead, 1)
	pb.l.Close()
}

// call this to find out if the server is dead.
func (pb *PBServer) isdead() bool {
	return atomic.LoadInt32(&pb.dead) != 0
}

// please do not change these two functions.
func (pb *PBServer) setunreliable(what bool) {
	if what {
		atomic.StoreInt32(&pb.unreliable, 1)
	} else {
		atomic.StoreInt32(&pb.unreliable, 0)
	}
}

func (pb *PBServer) isunreliable() bool {
	return atomic.LoadInt32(&pb.unreliable) != 0
}

func StartServer(vshost string, me string) *PBServer {
	pb := new(PBServer)
	pb.me = me
	pb.vs = viewservice.MakeClerk(me, vshost)
	// Your pb.* initializations here.
	pb.Viewnum = 0
	pb.Primary = ""
	pb.Backup = ""
	pb.data = make(map[string]string)
	pb.dataLocker = new(sync.RWMutex)
	pb.requestMap = make(map[int64]bool, 1024)

	rpcs := rpc.NewServer()
	rpcs.Register(pb)

	os.Remove(pb.me)
	l, e := net.Listen("unix", pb.me)
	if e != nil {
		log.Fatal("listen error: ", e)
	}
	pb.l = l

	// please do not change any of the following code,
	// or do anything to subvert it.

	go func() {
		for pb.isdead() == false {
			conn, err := pb.l.Accept()
			if err == nil && pb.isdead() == false {
				if pb.isunreliable() && (rand.Int63()%1000) < 100 {
					// discard the request.
					conn.Close()
				} else if pb.isunreliable() && (rand.Int63()%1000) < 200 {
					// process the request but force discard of reply.
					c1 := conn.(*net.UnixConn)
					f, _ := c1.File()
					err := syscall.Shutdown(int(f.Fd()), syscall.SHUT_WR)
					if err != nil {
						fmt.Printf("shutdown: %v\n", err)
					}
					go rpcs.ServeConn(conn)
				} else {
					go rpcs.ServeConn(conn)
				}
			} else if err == nil {
				conn.Close()
			}
			if err != nil && pb.isdead() == false {
				fmt.Printf("PBServer(%v) accept: %v\n", me, err.Error())
				pb.kill()
			}
		}
	}()

	go func() {
		for pb.isdead() == false {
			pb.tick()
			time.Sleep(viewservice.PingInterval)
		}
	}()

	return pb
}
